package com.atguigu.app.dwd.db;

import com.atguigu.util.MyKafkaUtil_hjy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author hjy
 * @create 2023/3/29 9:14
 */
public class Dwd_TradePayDetailSuc {
    public static void main(String[] args) {
        //todo 1 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall-flink/check");
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.setStateBackend(new HashMapStateBackend());
        //        System.setProperty("HADOOP_USER_NAME", "atguigu");
        //todo 2 从kafka获取数据
        tableEnv.executeSql(MyKafkaUtil_hjy.getTopicDBDDL("dwd_trade_pay_detail_suc"));

        //todo 3 建立支付状态表
        Table paymentInfoTable = tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['out_trade_no'] out_trade_no,\n" +
                "  `data`['order_id'] order_id,\n" +
                "  `data`['alipay_trade_no'] alipay_trade_no,\n" +
                "  `data`['total_amount'] total_amount,\n" +
                "  `data`['trade_body'] trade_body,\n" +
                "  `data`['payment_type'] payment_type,\n" +
                "  `data`['payment_status'] payment_status,\n" +
                "  `data`['create_time'] create_time,\n" +
                "  `data`['update_time'] update_time,\n" +
                "  `data`['callback_content'] callback_content,\n" +
                "  `data`['callback_time'] callback_time\n" +
                "from \n" +
                "  topic_db\n" +
                "where\n" +
                "`database` = 'edu'\n" +
                "and `table`='payment_info'\n" +
                "and `type`='insert'\n" +
                "and `data`['payment_status']='1602'\n"
                );
        tableEnv.createTemporaryView("payment_info",paymentInfoTable);
        //todo 4 建立订单表
        Table orderInfoTable = tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['origin_amount'] origin_amount,\n" +
                "  `data`['coupon_reduce'] coupon_reduce,\n" +
                "  `data`['final_amount'] final_amount,\n" +
                "  `data`['order_status'] order_status,\n" +
                "  `data`['out_trade_no'] out_trade_no,\n" +
                "  `data`['trade_body'] trade_body,\n" +
                "  `data`['session_id'] session_id,\n" +
                "  `data`['province_id'] province_id,\n" +
                "  `data`['create_time'] create_time,\n" +
                "  `data`['expire_time'] expire_time,\n" +
                "  `data`['update_time'] update_time\n" +
                "from\n" +
                "  topic_db\n" +
                "where\n" +
                "`database` = 'edu'\n" +
                "and `table`='order_info'\n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("order_info",orderInfoTable);

        //todo 5 建立订单明细表
        Table orderDetailTable = tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['course_id'] course_id,\n" +
                "  `data`['course_name'] course_name,\n" +
                "  `data`['order_id'] order_id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['origin_amount'] origin_amount,\n" +
                "  `data`['coupon_reduce'] coupon_reduce,\n" +
                "  `data`['final_amount'] final_amount,\n" +
                "  `data`['create_time'] create_time,\n" +
                "  `data`['update_time'] update_time,\n" +
                "  pt \n" +
                "from\n" +
                "  topic_db\n" +
                "where\n" +
                "`database` = 'edu'\n" +
                "and `table`='order_detail'\n" +
                "and `type`='insert'\n");
        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        //todo 7 三表关联
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "  od.id,\n" +
                "  od.course_id,\n" +
                "  od.course_name,\n" +
                "  od.order_id,\n" +
                "  od.user_id,\n" +
                "  od.origin_amount,\n" +
                "  od.coupon_reduce," +
                "  od.final_amount,\n" +
                "  od.create_time,\n" +
                "  od.update_time,\n" +
                "  oi.order_status,\n" +
                "  oi.out_trade_no,\n" +
                "  oi.trade_body,\n" +
                "  oi.session_id,\n" +
                "  oi.province_id,\n" +
                "  oi.expire_time,\n" +
                "  pi.alipay_trade_no,\n" +
                "  pi.total_amount,\n" +
                "  pi.payment_type,\n" +
                "  pi.payment_status,\n" +
                "  pi.callback_content,\n" +
                "  pi.callback_time\n" +
                "from \n" +
                "order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id=oi.id \n" +
                "join payment_info pi\n" +
                "on pi.order_id=oi.id \n" );
        tableEnv.createTemporaryView("result_table",resultTable);

        //todo 8 建立kafka主题
        tableEnv.executeSql("create table dwd_trade_pay_detail_suc(\n" +
                 "id STRING,\n" +
                 "course_id STRING,\n" +
                 "course_name STRING,\n" +
                 "order_id STRING,\n" +
                 "user_id STRING,\n" +
                 "origin_amount STRING,\n" +
                 "coupon_reduce STRING,\n" +
                 "final_amount STRING,\n" +
                 "create_time STRING,\n" +
                 "update_time STRING,\n" +
                 "order_status STRING,\n" +
                 "out_trade_no STRING,\n" +
                 "trade_body STRING,\n" +
                 "session_id STRING,\n" +
                 "province_id STRING,\n" +
                 "expire_time STRING,\n" +
                 "alipay_trade_no STRING,\n" +
                 "total_amount STRING,\n" +
                 "payment_type STRING,\n" +
                 "payment_status STRING,\n" +
                 "callback_content STRING,\n" +
                 "callback_time STRING"+
                ")"+ MyKafkaUtil_hjy.getKafkaSinkDDL("dwd_trade_pay_detail_suc"));
        //todo 9 输出数据
        tableEnv.executeSql("insert into dwd_trade_pay_detail_suc select * from result_table");
    }
}
